package dsx.msg;

import dsx.config.Config;
import dsx.config.ConfigItem;
import dsx.system.IThreadExWork;
import dsx.system.ThreadExPool;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;

public class MsgQueue {

    private static Queue<_Msg> msgQueue;

    private static ConcurrentHashMap<Integer, _MsgLinkedList> msgLinkedMap;

    private static ArrayList<MsgHandlerSet> msgHandlerSets;

    private static ThreadExPool threadExPool;

    private static String workType = "MsgQueue";

    private static int maxWorkerNum = 100;

    private static ArrayList<_MsgHandlerSetConf> confList;


    public static void init() {
        msgQueue = new LinkedList<>();
        msgLinkedMap = new ConcurrentHashMap<>();
        msgHandlerSets = new ArrayList<>();
        confList = new ArrayList<>();
        loadConf();
        for (_MsgHandlerSetConf conf : confList) {
            msgHandlerSets.add(new MsgHandlerSet(conf));
        }

        for (MsgHandlerSet msgHandlerSet : msgHandlerSets) {
            ArrayList<_MsgType> msgTypes = msgHandlerSet.getMsgTypes();
            for (_MsgType msgType: msgTypes) {
                if (!msgLinkedMap.containsKey(msgType.code)) {
                    msgLinkedMap.put(msgType.code, new _MsgLinkedList());
                }
                msgLinkedMap.get(msgType.code).addHandlerSet(msgHandlerSet);
                if (msgType.getMaxHandlingNum() > msgLinkedMap.get(msgType.code).maxRunningNum) {
                    msgLinkedMap.get(msgType.code).maxRunningNum = msgType.getMaxHandlingNum();
                }
            }
        }

        threadExPool = new ThreadExPool();
        threadExPool.setMaxWorkerNum(maxWorkerNum);
        threadExPool.addWork(workType, new _MsgHandler());
    }

    private static void loadConf() {
        try {
            int tmp = Integer.parseInt(Config.get("MsgQueue", "maxWorkerNum"));
            maxWorkerNum = tmp;
        } catch (Exception ex) {
        }

        try {
            ArrayList<ArrayList<ConfigItem>> itemsList = Config.getSections("MsgHandlerSet");
            for (ArrayList<ConfigItem> items : itemsList) {
                _MsgHandlerSetConf conf = new _MsgHandlerSetConf();
                for (ConfigItem item : items) {
                    switch (item.name) {
                        case "name":
                            conf.setName(item.value);
                            break;
                        case "path":
                            conf.setPath(item.value);
                            break;
                    }
                }
                confList.add(conf);
            }
        } catch (Exception ex) {
        }
    }

    public static boolean postMsg(int code, Object param) {
        synchronized (msgQueue) {
            if (msgQueue.offer(new _Msg(code, param))) {
                threadExPool.notify(workType);
                return true;
            } else {
                return false;
            }
        }
    }

    private static _Msg pullMsg() {
        synchronized (msgQueue) {
            if (msgQueue.isEmpty())
                return null;
            else
                return msgQueue.poll();
        }
    }

    public static void handleMsg() {
        while (true) {
            // 取一个消息
            _Msg msg = pullMsg();
            if (msg == null)
                break;
            // 取消息的处理句柄链
            _MsgLinkedList msgLinkedList = msgLinkedMap.get(msg.code);
            if (msgLinkedList == null)
                continue;
            msgLinkedList.handleMsg(msg);
        }
    }

    private static class _MsgHandler implements IThreadExWork {
        @Override
        public void run() {
            handleMsg();
        }
    }

    private static class _MsgLinkedList {

        int runningNum;
        int maxRunningNum;
        Queue<_Msg> msgQueue;
        List<MsgHandlerSet> handlerSets;

        public _MsgLinkedList() {
            runningNum = 0;
            maxRunningNum = 0;
            msgQueue = new LinkedList<>();
            handlerSets = new ArrayList<>();
        }

        public void handleMsg(_Msg msg) {
            // 第一段，
            synchronized (this) {
                if (runningNum >= maxRunningNum && maxRunningNum != 0) {
                    this.msgQueue.offer(msg);
                    return;
                } else {
                    runningNum++;
                }
            }
            // 第二段
            _Msg handlingMsg;
            synchronized (this) {
                if (this.msgQueue.isEmpty())
                    handlingMsg = msg;
                else
                    handlingMsg = this.msgQueue.poll();
            }
            handle(handlingMsg);
            // 第三段
            while (true) {
                synchronized (this) {
                    handlingMsg = this.msgQueue.poll();
                    if (handlingMsg == null)
                        break;
                    handle(handlingMsg);
                }
            }

            // 第四段
            synchronized (this) {
                runningNum--;
            }
        }

        private void handle(_Msg msg) {
            for (MsgHandlerSet handlerSet: handlerSets) {
                if (handlerSet.handleMsg(msg))
                    break;
            }
        }

        public void addHandlerSet(MsgHandlerSet msgHandlerSet) {
            this.handlerSets.add(msgHandlerSet);
        }

    }


}
